package com.seedog.mq;

import com.seedog.mq.model.Message;
import com.seedog.mq.core.MessageQueue;
import com.seedog.mq.exception.TopicNotFoundException;
import com.seedog.mq.exception.ProducerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 生产者实现类，提供向消息队列发送消息的功能
 */
public class ProducerImpl implements Producer {
    private static final Logger logger = LoggerFactory.getLogger(ProducerImpl.class);
    
    private final String producerId;
    private final MessageQueue messageQueue;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    
    public ProducerImpl(MessageQueue messageQueue) {
        if (messageQueue == null) {
            throw new IllegalArgumentException("MessageQueue cannot be null");
        }
        
        this.producerId = "producer-" + UUID.randomUUID().toString();
        this.messageQueue = messageQueue;
        logger.info("Producer created with ID: {}", producerId);
    }
    
    public ProducerImpl(MessageQueue messageQueue, String producerId) {
        if (messageQueue == null) {
            throw new IllegalArgumentException("MessageQueue cannot be null");
        }
        
        if (producerId == null || producerId.isEmpty()) {
            throw new IllegalArgumentException("Producer ID cannot be null or empty");
        }
        
        this.producerId = producerId;
        this.messageQueue = messageQueue;
        logger.info("Producer created with ID: {}", producerId);
    }
    
    @Override
    public String getProducerId() {
        return producerId;
    }
    
    @Override
    public String send(String topicName, Message message) throws TopicNotFoundException, ProducerException {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (message == null) {
            throw new IllegalArgumentException("Message cannot be null");
        }
        
        try {
            // 设置消息来源为当前生产者
            if (message.getProperties() != null) {
                message.getProperties().setSource(producerId);
            }
            
            String messageId = messageQueue.sendMessage(topicName, message);
            logger.debug("Producer {} sent message {} to topic {}", producerId, messageId, topicName);
            return messageId;
        } catch (TopicNotFoundException e) {
            throw e;
        } catch (Exception e) {
            logger.error("Failed to send message from producer {} to topic {}", producerId, topicName, e);
            throw new ProducerException("Failed to send message: " + e.getMessage(), e);
        }
    }
    
    @Override
    public String send(String topicName, String content) throws TopicNotFoundException, ProducerException {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (content == null) {
            throw new IllegalArgumentException("Content cannot be null");
        }
        
        Message message = new Message(topicName, content);
        return send(topicName, message);
    }
    
    @Override
    public String send(String topicName, String content, long expireTime) throws TopicNotFoundException, ProducerException {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (content == null) {
            throw new IllegalArgumentException("Content cannot be null");
        }
        
        if (expireTime <= 0) {
            throw new IllegalArgumentException("Expire time must be positive");
        }
        
        Message message = new Message(topicName, content);
        message.setExpireTime(expireTime);
        return send(topicName, message);
    }
    
    @Override
    public List<String> sendBatch(String topicName, List<Message> messages) throws TopicNotFoundException, ProducerException {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (messages == null || messages.isEmpty()) {
            throw new IllegalArgumentException("Messages cannot be null or empty");
        }
        
        try {
            // 设置所有消息的来源为当前生产者
            for (Message message : messages) {
                if (message.getProperties() != null) {
                    message.getProperties().setSource(producerId);
                }
            }
            
            List<String> messageIds = messageQueue.sendMessages(topicName, messages);
            logger.debug("Producer {} sent batch of {} messages to topic {}", 
                    producerId, messages.size(), topicName);
            return messageIds;
        } catch (TopicNotFoundException e) {
            throw e;
        } catch (Exception e) {
            logger.error("Failed to send batch messages from producer {} to topic {}", producerId, topicName, e);
            throw new ProducerException("Failed to send batch messages: " + e.getMessage(), e);
        }
    }
    
    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            logger.info("Producer {} closed", producerId);
        }
    }
    
    // 检查生产者是否已关闭
    private void checkClosed() {
        if (closed.get()) {
            throw new IllegalStateException("Producer '" + producerId + "' is closed");
        }
    }
}